/*
 * Copyright (c) 2016 moondev.cn
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cn.moondev.moonlog.example.moonlog.monitor.mq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

public class Consumer {
	
	private static final Logger log = LoggerFactory.getLogger(Consumer.class);
	
	/**
     * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例<br>
     * 注意：ConsumerGroupName需要由应用来保证唯一
     */
	private DefaultMQPushConsumer consumer;
	
	private String topic;
	
	
	public Consumer(String consumerGroupName,String namesrvAddr,String topic) {
		this.consumer = new DefaultMQPushConsumer(consumerGroupName);
		this.consumer.setNamesrvAddr(namesrvAddr);
		this.topic = topic;
	}
	
	public void start() {
		log.info("consumer start.......");
		if(!StringUtils.hasText(topic)) {
			return;
		}
		// 订阅指定topic下所有消息，注意：一个consumer对象可以订阅多个topic
	    try {
	    	log.info("consumer订阅主题：{}",topic);
			consumer.subscribe(topic, "*");
		} catch (MQClientException e) {
			throw new RuntimeException("日志系统消费者订阅主题失败：topic=" + topic);
		}
	    // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
        // 如果非第一次启动，那么按照上次消费的位置继续消费
	    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 设置为集群消费(区别于广播消费)
	    // 广播消费：一个consumer实例消费这个Topic对应的所有队列
	    // 集群消费：多个Consumer实例平均消费这个topic 对应的队列集合
	    // 注意：即使设置为集群模式，在集群模式下仍然不能防止消费端收到重复消息，需要消费端自己在业务中去重
	    consumer.setMessageModel(MessageModel.CLUSTERING);
	    // 设置批量处理，每次消费的条数小于等于100
	    consumer.setConsumeMessageBatchMaxSize(100);
	    consumer.registerMessageListener(new MessageListenerAdapter());
	    try {
			consumer.start();
		} catch (MQClientException e) {
			throw new RuntimeException("rocketmq consumer启动失败:" + e.getMessage());
		}
	}
	
	public void shutdown() {
		log.info("consumer shutdown.......");
		consumer.shutdown();
	}
}
